package com.yeming.flink.practice.transformation

import java.sql.{Connection, DriverManager, PreparedStatement, ResultSet}

import com.yeming.flink.practice.source.StationLog
import com.yeming.flink.practice.transformation.TranMapFunctionClass.getClass
import org.apache.flink.api.common.functions.RichMapFunction
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.scala._

object TranRickMapFunction {

  def main(args: Array[String]): Unit = {
    //加载环境
    val streamEnv: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    //读取数据源
    val fileFile: String = getClass.getResource("/station.log").getPath
    val stream: DataStream[String] = streamEnv.readTextFile(fileFile)
    //数据计算,将手机号码换成用户姓名
    val streamLog: DataStream[StationLog] = stream.map(line => {
      val arr: Array[String] = line.split(",")
      new StationLog(arr(0).trim, arr(1).trim, arr(2).trim, arr(3).trim, arr(4).trim.toLong, arr(5).trim.toLong)
    })





    streamLog.filter(log => log.callType.equals("success"))
      .map(new MyRichMapFunction()).print()

    //执行流计算
    streamEnv.execute()
//    //关闭连接
//    pst.close()
//    conn.close()
  }

  class MyRichMapFunction() extends RichMapFunction[StationLog, StationLog]{
    var conn:Connection =_
    var pst:PreparedStatement = _
    override def map(value: StationLog): StationLog = {
      println(getRuntimeContext.getTaskNameWithSubtasks)
      //查询主叫号码对应的姓名
      pst.setString(1,value.callOut)
      val result1: ResultSet = pst.executeQuery()
      if (result1.next()) {
        value.callOut = result1.getString(1)
      }
      //查询被叫号码对应的姓名
      pst.setString(1,value.callIn)
      val result2: ResultSet = pst.executeQuery()
      if (result2.next()) {
        value.callIn = result2.getString(1)
      }
      value
    }

    override def open(parameters: Configuration): Unit = {
      conn = DriverManager.getConnection("jdbc:mysql://localhost/mytest","root","root")
      pst = conn.prepareStatement("select name from t_mobile_info where mobile_num = ?")
    }

    override def close(): Unit = {
      pst.close()
      conn.close()
    }
  }
}
